package com.yanxu;

import com.yanxu.domain.Event2;
import com.yanxu.source.CustomSource2;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2025/7/27-2025
 * @Description: 这里用一句话描述这个类的作用
 */
public class Api_19_ProcessionFunction {
    public static void main(String[] args) {
        // 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // source 源
        DataStreamSource<Event2> dataSource = env.addSource(new CustomSource2());

        // 自定义 process 处理函数
        // 正常情况下，我们更多的是使用 map、flatmap、filter 这些封装好的 转换算子。
        // 但是，在一些比较复杂的情况下，我们可以使用 自定义 processFunction 算子，操作起来更加灵活。
        SingleOutputStreamOperator<Tuple2<String, Long>> processOperator = dataSource.process(new ProcessFunction<Event2, Tuple2<String, Long>>() {
            @Override
            public void processElement(Event2 event2,
                                       ProcessFunction<Event2, Tuple2<String, Long>>.Context ctx,
                                       Collector<Tuple2<String, Long>> collector) throws Exception {
                //ctx.timestamp();
                //使用自定义的 processFunction 可以获取更多的信息，比如，这里可以拿到 ctx 上下文信息等。
                String name = event2.getName();
                collector.collect(new Tuple2(name, 1));
            }
        });

        // 使用 keyBy 进行分区操作
        KeyedStream<Tuple2<String, Long>, String> keyedStream = processOperator.keyBy(r -> r.f0);

        // sum 聚合操作，并进行输出。
        keyedStream
                .sum("f1")
                .print();
    }
}